package com.atguigu.mapreduce.shiyan;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;


public class SortIntegers {
    /**
     * Mapper类，用于处理输入文件
      */
    public static class MapClass extends Mapper<Object, Text, IntWritable, NullWritable> {
        private IntWritable number = new IntWritable(); // 用于存储整数值
        // map函数，每个输入文件的每一行都会调用一次
        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString(); // 将输入的Text转换为字符串
            number.set(Integer.parseInt(line)); // 将字符串转换为整数
            context.write(number, NullWritable.get()); // 写入<整数, NullWritable>对，我们不关心值
        }
    }
    /**
     * Reducer类，用于对整数进行排序并输出排序位次
      */
    public static class Reduce extends Reducer<IntWritable, NullWritable, IntWritable, IntWritable> {
        private IntWritable rank = new IntWritable(); // 用于存储排序位次
        int sum = 0;
        // reduce函数，每个唯一的整数都会调用一次
        @Override
        public void reduce(IntWritable key, Iterable<NullWritable> values, Context context)
                throws IOException, InterruptedException {
            sum++;
            rank.set(sum); // 设置排序位次
            context.write(rank, key); // 写入<排序位次, 整数>对
        }
    }
    /**
     * Partitioner类，用于确保所有相同的整数都会被发送到同一个reducer
      */
    public static class PartitionerClass extends Partitioner<IntWritable, NullWritable> {
        // getPartition函数，用于决定哪个reducer应该接收当前的键值对
        @Override
        public int getPartition(IntWritable key, NullWritable value, int numPartitions) {
            return (key.get() & Integer.MAX_VALUE) % numPartitions; // 使用整数键的值对reducer的数量取模，以决定分区
        }
    }


    public static void main(String[] args) throws Exception {
        System.setProperty("HADOOP_USER_NAME","root");
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","hdfs://master:9000");
        Job job = Job.getInstance(conf, "sort integers");
        job.setJarByClass(SortIntegers.class);
        job.setMapperClass(MapClass.class);
        job.setReducerClass(Reduce.class);
        job.setPartitionerClass(PartitionerClass.class);
        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(NullWritable.class);
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(IntWritable.class);
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        job.setNumReduceTasks(1);
        // 设置输入和输出路径
        for (int i = 0; i < args.length - 1; ++i) {
            FileInputFormat.addInputPath(job, new Path(args[i]));
        }
        FileOutputFormat.setOutputPath(job, new Path(args[args.length - 1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
